package geektime.spring.reactor.simple;

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

@SpringBootApplication
@Slf4j
public class SimpleReactorDemoApplication implements ApplicationRunner {

	public static void main(String[] args) {
		SpringApplication.run(SimpleReactorDemoApplication.class, args);
	}

	@Override
	public void run(ApplicationArguments args) throws Exception {
//		test1();
//		test2();
//		test3();
//		test4();
		test5();
	}

	private void test1() throws Exception {
		Flux.range(1, 6)	// 创建 1 到 6 的序列
				.doOnRequest(n -> log.info("Request {}, {} number", Thread.currentThread(), n))		// 每次请求多少数据
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 1", Thread.currentThread()))	//
				.map(i -> {
					log.info("Publish {}, {}", Thread.currentThread(), i);
					return i;
				})
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 2", Thread.currentThread()))
				.subscribe(i -> log.info("Subscribe {}: {}", Thread.currentThread(), i),	// 对每个元素怎么消费
						e -> log.error("error {}", e.toString()),	// 发生错误如何处理
						() -> log.info("Subscriber {}, COMPLETE", Thread.currentThread())		// 运行完毕如何处理
				);
		Thread.sleep(2000);
	}

	private void test2() throws Exception {
		Flux.range(1, 6)
				.doOnRequest(n -> log.info("Request {}, {} number", Thread.currentThread(), n))
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 1", Thread.currentThread()))
				.publishOn(Schedulers.elastic())	// 之后的操作都在 elastic 线程上执行
				.map(i -> {
					log.info("Publish {}, {}", Thread.currentThread(), i);
					return i;
				})
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 2", Thread.currentThread()))
				.subscribeOn(Schedulers.single())	// publishOn 之前的操作 single 线程上执行
				.subscribe(i -> log.info("Subscribe {}: {}", Thread.currentThread(), i),
						e -> log.error("error {}", e.toString()),
						() -> log.info("Subscriber {}, COMPLETE", Thread.currentThread())
				);
		Thread.sleep(2000);
	}

	private void test3() throws Exception {
		Flux.range(1, 6)
				.doOnRequest(n -> log.info("Request {}, {} number", Thread.currentThread(), n))
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 1", Thread.currentThread()))
				.publishOn(Schedulers.elastic())
				.map(i -> {
					log.info("Publish {}, {}", Thread.currentThread(), i);
					return 10 / (i - 3);	// i 为 3 时抛出异常
				})
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 2", Thread.currentThread()))	// 由于会抛出异常，该代码未执行
				.subscribeOn(Schedulers.single())
				.onErrorReturn(-1)	// 有异常默认返回 -1，并终止执行
				.subscribe(i -> log.info("Subscribe {}: {}", Thread.currentThread(), i),
						e -> log.error("error {}", e.toString()),
						() -> log.info("Subscriber {}, COMPLETE", Thread.currentThread())
				);
		Thread.sleep(2000);
	}

	private void test4() throws Exception {
		Flux.range(1, 6)
				.doOnRequest(n -> log.info("Request {}, {} number", Thread.currentThread(), n))
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 1", Thread.currentThread()))
				.publishOn(Schedulers.elastic())
				.map(i -> {
					log.info("Publish {}, {}", Thread.currentThread(), i);
					return 10 / (i - 3);
				})
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 2", Thread.currentThread()))
				.subscribeOn(Schedulers.single())
				.onErrorResume(e -> {	// 打印异常信息
					log.error("Exception {}", e.toString());
					return Mono.just(-1);
				})
				.subscribe(i -> log.info("Subscribe {}: {}", Thread.currentThread(), i),
						e -> log.error("error {}", e.toString()),
						() -> log.info("Subscriber {}, COMPLETE", Thread.currentThread())
				);
		Thread.sleep(2000);
	}

	private void test5() throws Exception {
		Flux.range(1, 6)
				.publishOn(Schedulers.elastic())
				.doOnRequest(n -> log.info("Request {}, {} number", Thread.currentThread(), n))	// 放在 publishOn 下面，只拉取 4 个元素
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 1", Thread.currentThread()))	// 不执行
				.map(i -> {
					log.info("Publish {}, {}", Thread.currentThread(), i);
					return i;
				})
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 2", Thread.currentThread()))	// 不执行
				.subscribeOn(Schedulers.single())
				.subscribe(i -> log.info("Subscribe {}: {}", Thread.currentThread(), i),
						e -> log.error("error {}", e.toString()),
						() -> log.info("Subscriber {}, COMPLETE", Thread.currentThread()),	// 不执行
						s -> s.request(4)
				);
		Thread.sleep(2000);
	}

	private void test6() throws Exception {
		Flux.range(1, 6)
				.doOnRequest(n -> log.info("Request {}, {} number", Thread.currentThread(), n))	// 放在 publishOn 上面，拉取 256 个元素（虽然拉取 256 个元素但只发布处理 4 个）
				.publishOn(Schedulers.elastic())
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 1", Thread.currentThread()))	// 不执行
				.map(i -> {
					log.info("Publish {}, {}", Thread.currentThread(), i);
					return i;
				})
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 2", Thread.currentThread()))	// 不执行
				.subscribeOn(Schedulers.single())
				.subscribe(i -> log.info("Subscribe {}: {}", Thread.currentThread(), i),
						e -> log.error("error {}", e.toString()),
						() -> log.info("Subscriber {}, COMPLETE", Thread.currentThread()),	// 不执行
						s -> s.request(4)
				);
		Thread.sleep(2000);
	}

	private void test7() throws Exception {
		Flux.range(1, 6)
				.doOnRequest(n -> log.info("Request {}, {} number", Thread.currentThread(), n))	// 放在 publishOn 上面，拉取 256 个元素（虽然拉取 256 个元素但只发布处理 4 个）
				.publishOn(Schedulers.elastic())
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 1", Thread.currentThread()))	// 不执行
				.map(i -> {
					log.info("Publish {}, {}", Thread.currentThread(), i);
					return 10 / (i - 3);
				})
				.doOnComplete(() -> log.info("Publisher {}, COMPLETE 2", Thread.currentThread()))	// 不执行
				.subscribeOn(Schedulers.single())
				.onErrorResume(e -> {
					log.error("Exception {}", e.toString());
					return Mono.just(-1);
				})
				.subscribe(i -> log.info("Subscribe {}: {}", Thread.currentThread(), i),
						e -> log.error("error {}", e.toString()),
						() -> log.info("Subscriber {}, COMPLETE", Thread.currentThread()),	// 执行(有异常抛出，并有异常拦截所以执行)
						s -> s.request(4)
				);
		Thread.sleep(2000);
	}
}

